package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.message.CheckResultMessage;
import cn.tedu.charging.common.pojo.message.ProgressMessage;
import cn.tedu.charging.order.service.ConsumerService;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 回调函数类,需要重建成容器bean对象并且交给绑定一个mqttClient使用
 * 专门负责监听这个mqttClient的一举一动
 * 1. 连接成功
 * 2. 发送消息
 * 3. 断开连接
 * 4. 收到消息
 */
@Component
@Slf4j
public class MqttClientCallback implements MqttCallbackExtended {
    @Autowired
    private RedisTemplate redisTemplate;
    /*
        方法作用: 当mqttClient调用connect并且成功和emqx建立通信之后立即调用
        参数:
        boolean reconnect: 如果本次建立的连接是重新连接,reconnect就是true 第一次连接false
        String serverURI: 连接服务端address
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if(reconnect){
            log.debug("重新连接成功,serverURI={}",serverURI);
        }else{
            log.debug("首次连接成功,serverURI={}",serverURI);
        }
    }
   /*
       方法作用: 当mqttClient调用disconnect方法断开连接后,或者由于网络波动,导致和服务端连接通信超时失败,
       回调就会调用这个方法
       参数:
       Throwable cause: 断开连接的原因
    */
    @Override
    public void connectionLost(Throwable cause) {
        log.debug("连接丢失,原因={}",cause);
    }
    /*
       方法作用: 当mqttClient发送消息的时候,只要消息调用publish结束且未发生异常,不管emqx是否能够接收到
       反馈接收成功还是失败,都会调用这个deliveryComplete,处理发送完成的业务
       qos服务质量级别中0 表示根本不关心投递发送结果,所以,如果客户端mqttClient使用qos=0级别发送消息,不会有任何异常
       如果要是1和2级别,可以从token中获取反馈结果,如果emqx反馈是失败的,会有个异常信息
       IMqttDeliveryToken token: 发送的消息对象 映射,包含了你发送消息的所有内容,以及双发通信的结果
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息发送已经完成");
        //读取token数据 判断消息投递结果
        boolean complete = token.isComplete();
        //true表示发到了服务端,false根本没发到
        if (complete){
            log.debug("消息成功到了服务端进程");
            MqttException exception = token.getException();//如果有异常说明服务端没有成功接收或者存储失败
            if (exception!=null){
                log.debug("服务端处理接收消息的时候失败,失败原因:{}", exception);
            }else{
                log.debug("消息发送成功,且服务端接收处理也成功");
            }
        }else{
            log.debug("消息没有成功发到服务端");
        }
    }
    /*
        这个方法是消费的入口
        方法作用: 当mqttClient连接对象订阅了某个主题,一旦这个主题中出现消息,就会通过这个方法将消息传递给客户端
        参数:
        String topic: 消息来源的topic名称 客户端订阅了哪些主题 消息来源就可能是其中一个
        MqttMessage message: 消息对象,包含了消息的所有内容
     */
    @Autowired
    private ConsumerService consumerService;
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("收到消息,topic={},message={}",topic,message);
        //1.解析消息数据内容 转化成CheckResultMessage 只有满足主题名称是charging/device/check/result
        if (topic!=null&&topic.equals("charging/device/check/result")){
            String json=message.toString();
            CheckResultMessage msg = JSON.parseObject(json, CheckResultMessage.class);
            //1.做一些抢锁的准备,setnx的操作对象,锁的key值要绑定订单编号
            ValueOperations valueOps = redisTemplate.opsForValue();
            String lockKey="charging:order:"+msg.getBillId()+".lock";//锁的key值
            //2.为循环准备一些参数
            boolean haveLock=false;//抢锁结果
            int count=0;//重抢的次数
            try {
                do {
                    //判断一下抢的次数 第一次count=0 重抢三次count>0 count<4 睡200毫秒 重抢三次以上 count>=4
                    if (count > 0 && count < 4) {
                        //说明是重抢,需要等待200毫秒
                        Thread.sleep(200);
                    } else if (count >= 4) {
                        return;//方法执行结束 不抢锁,也不执行业务了;
                    }
                    //3.开始抢锁 setnx key value EX 5
                    haveLock = valueOps.setIfAbsent(lockKey, "", 5, TimeUnit.SECONDS);
                    count++;
                } while (!haveLock);
                //消费核心
                consumerService.handleCheckResult(msg);
            }catch (Exception e){
                log.error("处理检查结果消息异常:{}",e);
            }finally {
                //5.释放锁
                if (haveLock) {
                    redisTemplate.delete(lockKey);
                }
            }
        }else if (topic!=null&&topic.equals("charging/device/progress")){
            //消费接收设备同步充电进度消息 拿到消息内容
            String json=message.toString();
            //转化成消息封装对象
            ProgressMessage msg=JSON.parseObject(json,ProgressMessage.class);
            //交给业务层处理业务逻辑
            consumerService.handleChargingProgress(msg);
        }
    }


}
